package com.demo.rocketmq.consumer;

import com.demo.rocketmq.config.ConsumerConfig;
import com.demo.rocketmq.config.RocketMqConfig;
import com.demo.rocketmq.config.TopicConfig;
import com.demo.rocketmq.listener.DemoSequenceConsumerListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author wangfengchen
 */
@Configuration
@Slf4j
public class DemoSequenceConsumer {

  @Value("${rocketmq.consumer.sequenceConsumerGroup:sequenceConsumerGroup}")
  private String group;
  @Autowired
  private RocketMqConfig rocketMqConfig;
  @Autowired
  private ConsumerConfig consumerConfig;
  @Autowired
  private TopicConfig topicConfig;
  @Autowired
  private DemoSequenceConsumerListener demoSequenceConsumerListener;


  @Bean(initMethod = "start", destroyMethod = "shutdown")
  public DefaultMQPushConsumer buildDemoSequenceConsumer() throws Exception {

    DefaultMQPushConsumer consumer;
    if (rocketMqConfig.getAclActive()) {
      // 生产环境使用阿里云mq
      consumer = new DefaultMQPushConsumer(group, rocketMqConfig.getAclRPCHook(), new AllocateMessageQueueAveragely());
      // 设置使用接入方式为阿里云，在使用云上消息轨迹的时候，需要设置此项，如果不开启消息轨迹功能，则运行不设置此项.
      consumer.setAccessChannel(AccessChannel.CLOUD);
    } else {
      consumer = new DefaultMQPushConsumer(group);
    }

    consumer.setNamesrvAddr(rocketMqConfig.getNamesrvAddr());
    // 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
    // 如果非第一次启动，那么按照上次消费的位置继续消费
    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    // 设置消费模型，集群还是广播，默认为集群
    consumer.setMessageModel(MessageModel.CLUSTERING);
    // 设置一次消费消息的条数，默认为1条
    consumer.setConsumeMessageBatchMaxSize(consumerConfig.getConsumeMessageBatchMaxSize());
    // 只能注册一个，最后的生效，所以需要多个生产者实例
    consumer.registerMessageListener(demoSequenceConsumerListener);
    try {
      // 设置该消费者订阅的主题和tag，如果是订阅该主题下的所有tag，则tag使用*，例如：demoTopic~*
      // 如果需要指定订阅该主题下的某些tag，则使用||分割，例如topic~tag1||tag2||tag3;topic2~*;
      String[] topicTagsArr = topicConfig.getSequenceTopic().split(";");
      for (String topicTags : topicTagsArr) {
        String[] topicTag = topicTags.split("~");
        if (topicTag.length > 1) {
          consumer.subscribe(topicTag[0], topicTag[1]);
        } else {
          consumer.subscribe(topicTag[0], "*");
        }
      }
    } catch (MQClientException e) {
      log.error("顺序消费者启动失败");
      throw new Exception(e);
    }
    log.info("《====================【顺序消费者启动成功】====================》");
    return consumer;
  }

}